package com.burges.net.dataStream.windows.windowAssigner

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * 滚动窗口基本使用
  * 默认窗口时间的时区是UTC-0，因此使用的时候需要设定偏移量调整时区，国内需要设定Time.hours(-8)
  */
object TumbliingWindow {

  def main(args: Array[String]): Unit = {

    case class WebLog(id: Long, content: String, trackTime: Long)

    val parameterTool = ParameterTool.fromArgs(args)

    //定义时间类型为EventTime
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val inputStream = env.fromElements(WebLog(2, "flink", 21L), WebLog(4, "flink", 1L), WebLog(3, "flink", 4L))

    //定义EventTimeTumblingWindows
    val tumblingEventTimeWindows = inputStream.keyBy(_.id)
                    // 通过使用TumblingEventTimeWindows定义EventTIme滚动窗口
                    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    //定义窗口函数
//                    .process(...)

    // 定义ProcessTime Tumbling Windows
    val tumblingProcessingTimeWindows = inputStream.keyBy(_.id)
            //通过使用TumblingProcessingTimeWindows定义ProcessTime滚动窗口
                    .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    //定义窗口函数
//                    .process(....)

    //定义的事件类型根据用户实现设定的时间概念确定
    inputStream.keyBy(_.id)
            //通过使用timeWindow方式定义滚动窗口，窗口事件类型根据time characteristic确定
            .timeWindow(Time.seconds(1))
//            .process(...)

  }

}
